package com.zws.cucumber.optimization.socket.oio;

import com.zws.cucumber.optimization.socket.connector.ZwsConnectorAdaptor;
import org.eclipse.jetty.util.BufferUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.util.concurrent.TimeUnit;

import static com.zws.cucumber.optimization.socket.oio.ZwsSocketFaultEnum.*;

public class ZwsSocketInputStream extends InputStream {

    private static final Logger logger = LoggerFactory.getLogger(ZwsSocketInputStream.class);

    private ZwsSocketOutputStream outputStream;

    private boolean first = true;
    private ZwsSocketFaultEnum fault = NO_FAULT;
    private ZwsConnectorAdaptor connectorAdaptor;
    private ByteBuffer inputBuffer;

    public ZwsSocketInputStream(ZwsConnectorAdaptor connectorAdaptor) {
        this.connectorAdaptor = connectorAdaptor;
    }

    @Override
    public int read() throws IOException {
//        if (logger.isTraceEnabled()) {
//            logger.trace("read(), ByteBuffer: {}", System.identityHashCode(outputBuffer));
//        }
        handleIfFirstRead();
        throwExceptionIfNeeded();
        return inputBuffer.get() & 0xFF;
    }

    @Override
    public int read(byte[] b, int off, int len) throws IOException {
//        if (logger.isTraceEnabled()) {
//            logger.trace("read(byte[], int, int), ByteBuffer: {}", System.identityHashCode(outputBuffer));
//        }
        handleIfFirstRead();
        throwExceptionIfNeeded();
        if (inputBuffer.hasRemaining()) {
//            logger.debug("read(byte[], int, int), has remaining");
            int s = inputBuffer.position();
            inputBuffer.get(b, off, Math.min(len, inputBuffer.remaining()));
            return inputBuffer.position() - s;
        } else {
//            logger.debug("read(byte[], int, int), no remaining");
            return 0;
        }
    }

    @Override
    public int available() {
        return inputBuffer.remaining();
    }

    private void handleIfFirstRead() throws IOException {
        if (first) {
            first = false;
            try {
                ByteBuffer outputBuffer = outputStream.getOutputBuffer();
                outputBuffer.flip();
                if (logger.isTraceEnabled()) {
                    logger.trace("handleIfFirstRead, send: {}", BufferUtil.toUTF8String(outputBuffer));
                }
                long s = System.nanoTime();
                Object result = connectorAdaptor.getResponse(outputBuffer, false, 30, TimeUnit.SECONDS);
                logger.debug("local connector send/recv, {}ns", System.nanoTime() - s);
                if (result instanceof ByteBuffer b) {
                    inputBuffer = b;
                    if (logger.isTraceEnabled()) {
                        logger.trace("handleIfFirstRead, recv: {}", BufferUtil.toUTF8String(inputBuffer));
                    }

                } else if (result == ZwsSocketFaultEnum.CONNECTION_RESET) {
                    if (logger.isTraceEnabled()) {
                        logger.trace("handleIfFirstRead, SocketException, Connection reset");
                    }
                    setFault(CONNECTION_RESET);

                } else {
                    if (logger.isTraceEnabled()) {
                        logger.trace("handleIfFirstRead, SocketTimeoutException");
                    }
                    setFault(READ_TIMEOUT);

                }

                outputStream.resetState();
            } catch (IOException e) {
                logger.error("handleIfFirstRead, IOException", e);
                throw e;
            } catch (Exception e) {
                logger.error("handleIfFirstRead, Exception", e);
                throw new RuntimeException(e);
            }
        }
    }

    private void throwExceptionIfNeeded() throws IOException {
        switch (fault) {
            case READ_TIMEOUT -> throw new SocketTimeoutException("Read timed out");
            case CONNECTION_RESET -> throw new SocketException("Connection reset");
        }
    }

    public void set(ByteBuffer resp) {
        inputBuffer = resp;
    }

    public void setOutputStream(ZwsSocketOutputStream outputStream) {
        this.outputStream = outputStream;
    }

    public void setFault(ZwsSocketFaultEnum fault) {
        this.fault = fault;
    }

    public void resetState() {
        first = true;
        fault = NO_FAULT;
        inputBuffer = null;
    }
}
